package com.bw.sparkstreaming.job6

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * hadoop,hadoop,spark,hive，？
  * hadoop,2
  * spark,1
  * hive,1
  * 单词计数的时候进行黑名单过滤
  * 比如单词计数的时候，我们规定 ？ ！ *
  * 这些特殊的符号就不参与单词计数了。
  *
  * 实现了一个黑名单过滤的这样的一个效果。
  *
  */
object TransformTest {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)

    val sparkConf = new SparkConf().setAppName("transformtest").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(2))

    val lines = ssc.socketTextStream("192.168.134.130", 8888)
    val words = lines.flatMap(_.split(","))
    val wordsDStream = words.map(x => (x, 1))
    //我们这儿自己创建一个黑名单
    //但是如果是在生产中的话，这个黑名单是从我们数据库（reids，MySQL）里面读取出来的。
    //还有一种可能就是这个黑名单还是动态的。
    val filterRDD: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(List("?", "!", "&"))
      .map((_, true))

    //广播变量
    val filterBroadCast = ssc.sparkContext.broadcast(filterRDD.collect())

    /**
      * 大家主要，这个transformation方便的地方在于
      * 1. 我们可以把DStream转换成为RDD进行处理(sparkcore的编程就可以使用上了)
      * 2. 如果我们可以把一个东西变成RDD，我们是不是也可以把RDD转换成为DataFrame/DataSet
      *    我们是不是可以就可以使用SparkSQL的编程 了吧？
      *    其实我们就可以写SQL语句了？
      *    我们是不是可以 SparkStreaming => SparkSQL,SparkCore的编程
      */
    //dstream -> rdd
    val wordAndOneDStream: DStream[(String, Int)] = wordsDStream.transform(rdd => {
      //使用到了外部的变量
      val filterRDD: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(filterBroadCast.value)


      /**
        * hadoop,1  join  ?,true => join不上
        *
        * ?,1  join ?,true => ("?",{1,true}) 能join上的，我们就是要过滤能join上的数据。
        * 因为我们用的是左外连接，所以如果能join上 Option true
        * 如果join不上，那么这个Option 值是NONE
        */
      val result: RDD[(String, (Int, Option[Boolean]))] = rdd.leftOuterJoin(filterRDD)

      /**
        * filter过滤：
        * 如果我们需要的数据，返回 true
        * 不需要的数据，返回false
        */
      val filterResultRDD: RDD[(String, (Int, Option[Boolean]))] = result.filter(tuple => {
        val word = tuple._1
        val joinResult = tuple._2
        //没有join上的数据，是不是就不属于黑名单
        //就是我们需要统计的单词
        //在scala编程我们最后一行代码就是返回值，所以不要加return
         joinResult._2.isEmpty
      })
      filterResultRDD.map(tuple => (tuple._1, 1))
    })

    val result = wordAndOneDStream.reduceByKey(_+_)
    result.print()
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()



  }

}

